package com.gitee.dufafei.spark.connector.db

import java.sql.{Connection, DriverManager}

import com.gitee.dufafei.spark.pattern.{Borrow, Logging}
import javax.sql.DataSource
import scalikejdbc.{ConnectionPool, ConnectionPoolSettings, DB, DataSourceConnectionPool}

/**
 * <dependency>
 * <groupId>org.scalikejdbc</groupId>
 * <artifactId>scalikejdbc_2.11</artifactId>
 * <version>3.4.1</version>
 * </dependency>
 */
object DbClient extends Borrow with Logging with Serializable {

  /**
   * 这里使用默认连接池-c3p0
   */
  def initPool(name: String,
               url: String,
               username: String,
               password: String,
               settings: ConnectionPoolSettings): Unit = {
    if(!ConnectionPool.isInitialized(Symbol(name))) {
      ConnectionPool.add(Symbol(name), url, username, password, settings)
      info(s"successful initialization the database connection pool: $name")
    }
  }

  def initPool(name: String, dataSource: DataSource): Unit = {
    if(!ConnectionPool.isInitialized(Symbol(name))) {
      ConnectionPool.add(Symbol(name), new DataSourceConnectionPool(dataSource))
      info(s"successful initialization the database connection pool: $name")
    }
  }

  def getConnection(driverClassName: String,
                    url: String,
                    username: String,
                    password: String): Connection = {
    Class.forName(driverClassName)
    DriverManager.getConnection(url, username, password)
  }

  def usingDB[A](conn: Connection)(execute: DB => A): A = {
    val db = DB(conn)
    db.autoClose(false)
    using(db)(execute)
  }

  def usingDB[A](name: String)(execute: DB => A): A = {
    val db = DB(ConnectionPool(Symbol(name)).borrow())
    db.autoClose(false)
    using(db)(execute)
  }
}
